package com.settlement.system.service.impl;

import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.settlement.system.common.constant.KafkaConstants;
import com.settlement.system.common.constant.SystemConstants;
import com.settlement.system.common.util.DateUtils;
import com.settlement.system.common.util.NumberUtils;
import com.settlement.system.common.util.StringUtils;
import com.settlement.system.handler.disruptor.MessageVo;
import com.settlement.system.mapper.XhsSaleDetailMapper;
import com.settlement.system.model.entity.KafkaSendErrorRecord;
import com.settlement.system.model.entity.SysImportErrorDetail;
import com.settlement.system.model.entity.SysStore;
import com.settlement.system.model.entity.XhsSaleDetail;
import com.settlement.system.model.form.XhsSaleDetailForm;
import com.settlement.system.model.query.XhsSaleDetailQuery;
import com.settlement.system.model.vo.XhsSaleDetailImportVo;
import com.settlement.system.service.*;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * <p>
 * 小红书货款明细 服务实现类
 * </p>
 *
 * @author huangrensen
 * @since 2024-07-02
 */
@Service
@RequiredArgsConstructor
public class XhsSaleDetailServiceImpl extends ServiceImpl<XhsSaleDetailMapper, XhsSaleDetail> implements XhsSaleDetailService {

    private final Logger logger = LoggerFactory.getLogger(XhsSaleDetailServiceImpl.class);
    private final SysStoreService sysStoreService;

    private final RedisTemplate redisTemplate;

    private final SysImportErrorDetailService importErrorDetailService;

    private final KafkaTemplate<String, String> kafkaTemplate;

    private final KafkaSendErrorRecordService kafkaSendErrorRecordService;
    private final SnowflakeIdWorker snowflakeIdWorker = SnowflakeIdWorker.getInstance();

    @Override
    public Page<XhsSaleDetail> getPage(XhsSaleDetailQuery queryParams) {
        // 查询参数
        int pageNum = queryParams.getPageNum();
        int pageSize = queryParams.getPageSize();
        // 拼接月份列表
        List<String> monthList = new ArrayList<>();
        String endMonthStr = queryParams.getEndMonth().replaceAll("-", "");
        LocalDate endMonth = LocalDate.parse(endMonthStr + "01", DateTimeFormatter.ofPattern("yyyyMMdd"));
        LocalDate startTime = LocalDate.parse(queryParams.getStartMonth().replaceAll("-", "") + "01", DateTimeFormatter.ofPattern(
                "yyyyMMdd"));
        // 获取已经生成的表名
        Object object = redisTemplate.opsForValue().get(SystemConstants.XHS_SALE_DETAIL);
        List<String> tableNameList;
        if (object != null) {
            tableNameList = (List<String>) object;
        } else {
            tableNameList = this.baseMapper.getAllTable();
            // 添加缓存
            redisTemplate.opsForValue().set(SystemConstants.XHS_SALE_DETAIL, tableNameList, 1, TimeUnit.HOURS);
        }
        // 循环追加月份
        while (!startTime.isAfter(endMonth)) {
            monthList.add(startTime.format(DateTimeFormatter.ofPattern("yyyyMM")));
            startTime = startTime.plusMonths(1);
        }
        // 根据月份判断数据库表是否存在
        List<String> existTableList = new ArrayList<>();
        String tableName;
        for (String month : monthList) {
            tableName = "xhs_sale_detail_%s".formatted(month);
            if (tableNameList.contains(tableName)) {
                existTableList.add(tableName);
            }
        }
        // 查询数据
        Page<XhsSaleDetail> page = new Page<>(pageNum, pageSize);
        if (existTableList.size() == 0) {
            return page;
        }
        queryParams.setTableNameList(existTableList);
        Page<XhsSaleDetail> mapperPage = this.baseMapper.getPage(page, queryParams);
        if (mapperPage.getRecords().size() > 0) {
            SysStore store;
            for (XhsSaleDetail record : mapperPage.getRecords()) {
                store = sysStoreService.getById(record.getStoreId());
                if (Objects.nonNull(store)) {
                    record.setStoreName(store.getCommercePlatformStoreName());
                }
            }
        }
        return mapperPage;
    }

    @Override
    public boolean saveXhsSaleDetail(XhsSaleDetailForm xhsSaleDetailForm) {
        XhsSaleDetail xhsSaleDetail = new XhsSaleDetail();
        BeanUtils.copyProperties(xhsSaleDetailForm, xhsSaleDetail);
        xhsSaleDetail.setCreateTime(null);
        return this.saveOrUpdate(xhsSaleDetail);
    }

    @Override
    public XhsSaleDetailForm getXhsSaleDetailFormData(Long id) {
        XhsSaleDetail xhsSaleDetail = this.getById(id);
        XhsSaleDetailForm xhsSaleDetailForm = new XhsSaleDetailForm();
        BeanUtils.copyProperties(xhsSaleDetail, xhsSaleDetailForm);
        return xhsSaleDetailForm;
    }

    @Override
    public boolean deleteXhsSaleDetail(String ids) {
        Assert.isTrue(StringUtils.hasText(ids), "删除的数据为空");
        List<Long> idArray = Arrays.stream(ids.split(",")).map(Long::parseLong).collect(Collectors.toList());
        return this.removeByIds(idArray);
    }

    @KafkaListener(id = KafkaConstants.HWG_TOPIC_XHS_SALE_DETAIL, topics = KafkaConstants.HWG_TOPIC_XHS_SALE_DETAIL, groupId = "yzConsumerGroup")
    public void importData(List<ConsumerRecord<String, String>> records, Acknowledgment acknowledgment) {
        logger.info("【{}】当前消费者本次拉取的数据量为：{}", KafkaConstants.HWG_TOPIC_XHS_SALE_DETAIL, (Objects.isNull(records) ? 0 : records.size()));
        if (Objects.isNull(records) || records.isEmpty()) {
            logger.error("【{}】无消费数据...", KafkaConstants.HWG_TOPIC_XHS_SALE_DETAIL);
        }
        String orderNo = "";
        for (ConsumerRecord<String, String> record : records) {
            String importVoStr = record.value();
            try {
                XhsSaleDetailImportVo importVo = JSONObject.parseObject(importVoStr, XhsSaleDetailImportVo.class);
                orderNo = importVo.getBusinessNo();
                this.importData(Collections.singletonList(importVo));
            } catch (Exception e) {
                logger.info("消费失败：{}", e.getMessage());
                // 保存失败数据
                SysImportErrorDetail detail = new SysImportErrorDetail();
                detail.setPlatformAndType("小红书货款明细");
                detail.setDataText(importVoStr);
                detail.setOrderNo(orderNo);
                detail.setQueueName(KafkaConstants.HWG_TOPIC_XHS_SALE_DETAIL);
                detail.setRemark(e.getMessage());
                importErrorDetailService.checkAndCreate(detail);
            }
        }
        // 提交消费位点
        acknowledgment.acknowledge();
    }

    @Override
    public void importData(List<XhsSaleDetailImportVo> voList) {
        redisTemplate.delete(SystemConstants.XHS_SALE_DETAIL);
        List<String> billMonthList = voList.stream().map(XhsSaleDetailImportVo::getBillMonth).distinct().toList();
        XhsSaleDetail xhsSaleDetail;
        for (String billMonth : billMonthList) {
            if (!StringUtils.hasText(this.baseMapper.existTable("xhs_sale_detail_%s".formatted(billMonth)))) {
                this.baseMapper.createTable("xhs_sale_detail_%s".formatted(billMonth));
            }
        }
        List<XhsSaleDetail> xhsSaleDetailList = new ArrayList<>();
        for (XhsSaleDetailImportVo importVo : voList) {
            xhsSaleDetail = new XhsSaleDetail();
            BeanUtils.copyProperties(importVo, xhsSaleDetail);
            xhsSaleDetail.setStoreId(importVo.getStoreId());
            xhsSaleDetail.setId(snowflakeIdWorker.nextId(importVo.getBillMonth()));
            xhsSaleDetail.setTableName("xhs_sale_detail_%s".formatted(importVo.getBillMonth()));
            // 创建时间
            xhsSaleDetail.setBillCreateTime(DateUtils.convertStringToLocalDateTime(importVo.getBillCreateTime()));
            // 收入（元）
            xhsSaleDetail.setIncomeAmount(NumberUtils.convertBigDecimal(importVo.getIncomeAmount()));
            // 支出（元）
            xhsSaleDetail.setPayAmount(NumberUtils.convertBigDecimal(importVo.getPayAmount()));
            // 账户余额（元）
            xhsSaleDetail.setBalanceAmount(NumberUtils.convertBigDecimal(importVo.getBalanceAmount()));
            xhsSaleDetailList.add(xhsSaleDetail);
        }
        if (xhsSaleDetailList.size() > 0) {
            xhsSaleDetailList.forEach(this.baseMapper::insertData);
        }
        // 发送队列消息通知
        MessageVo vo;
        for (XhsSaleDetail item : xhsSaleDetailList) {
            vo = new MessageVo();
            vo.setDataId(item.getId());
            vo.setStoreId(item.getStoreId());
            vo.setBillMonth(item.getBillMonth());
            // 根据字典中的备注查找对应的数据库表字段名
            vo.setTableNamePrefix("xhs_sale_detail_");
            // 从字段管理里找到对应的字段类型编码，获取对应的billType
            vo.setDictCode("redBookSaleDetail");
            // 推送到转换标准账单的消息队列
            vo.setUuid(snowflakeIdWorker.nextId());
            CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_BILL_QUEUE, JSONObject.toJSONString(vo));
            MessageVo finalVo = vo;
            completableFuture.whenComplete((sendResult, throwable) -> {
                if(Objects.nonNull(throwable)){
                    throwable.printStackTrace();
                    KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                    errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                    errorRecord.setRemark(throwable.getMessage());
                    errorRecord.setUuid(finalVo.getUuid());
                    errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_BILL_QUEUE);
                    kafkaSendErrorRecordService.checkAndSave(errorRecord);
                }
            });
        }
    }

    @Override
    public List<XhsSaleDetailImportVo> findByIds(String idsStr) {
        if (!StringUtils.hasText(idsStr)) {
            return new ArrayList<>();
        }
        List<Long> ids = Arrays.stream(idsStr.split(",")).map(Long::parseLong).collect(Collectors.toList());
        List<XhsSaleDetail> xhsSaleDetailList = this.getBaseMapper().selectBatchIds(ids);
        List<XhsSaleDetailImportVo> voList = new ArrayList<>();
        for (XhsSaleDetail xhsSaleDetail : xhsSaleDetailList) {
            XhsSaleDetailImportVo vo = new XhsSaleDetailImportVo();
            BeanUtils.copyProperties(xhsSaleDetail, vo);
            voList.add(vo);
        }
        return voList;
    }
}
